%%% File    : net_server.erl
%%% Author  :  <whhbest>
%%% Description : 
%%% Created : 13 Mar 2011 by  <>

-module(net_server).

-behaviour(gen_server).

-export([start/2]).

-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
	terminate/2, code_change/3]).

-export([acceptor/2, loop/1]).

-define(Size, 10).

%%-define(Maxcon, 5000).

-define(SERVER, ?MODULE).

-record(state, {sup,
	       ip,
	       port,
	       lsock,
	       pool,
	       connect,
	       connum,
	       flash}).

%%------------------
start(Sup, Port) ->
    gen_server:start({local, ?SERVER}, ?MODULE, [Sup, Port], []).

%%------------
init([Sup, Port]) ->
    process_flag(trap_exit, true),
    Opts = [binary,
	   {packet, 2},
	   {reuseaddr, true},
	   {backlog, 100},
	   {active, false},
	   {nodelay, true}],
    %%{ok, Ip} = get_local_ip(),
    {Tag, Lsock} = gen_tcp:listen(Port, Opts),
    case Tag of
	error ->
	    exit(Lsock);
	ok ->
	    Fun = fun(_, List) ->
			  Pid = spawn_link(?MODULE, acceptor, [self(), Lsock]),
			  [Pid|List]
		  end,
	    Pool = start_pool(Fun),
	    case flash_server:start_link(self()) of
		{ok, Flash} ->
		     {ok, #state{sup=Sup,
				 ip={169,254,104,75},
				 port=Port,
				 lsock=Lsock,
				 pool=Pool,
				 connum=0,
				 connect=[],
				flash=Flash}};
		Err ->
		    {stop, {error, Err}}
             end
     end.

handle_call('WHERE AND COUNT', _From, S=#state{ip=Ip,
					       port=Port,
					       connum=Num}) ->
    {reply, {Ip, Port, Num}, S};

handle_call(_Req, _From, S) ->
    {reply, ok, S}.

handle_cast(redirect, S) ->
    redirect(S);

handle_cast({do_exit, Pid}, S=#state{sup=Sup}) ->
    case Pid =:= Sup of
	true ->
	    {stop, normal, S};
	false ->
	    {noreply, S}
    end;

handle_cast({acceptor, {Tag, Pid}}, S=#state{lsock=Lsock,
					     pool=Pool,
					     connum=Num,
					     connect=Conn}) ->
    Npid = spawn_link(?MODULE, acceptor, [self(), Lsock]),
    case Tag of
	ok ->
	    {noreply, S#state{pool=[Npid|lists:delete(Pid, Pool)],
			     connum=Num+1,
			     connect=[Pid|Conn]}};
	error ->
	    {noreply, S#state{pool=[Npid|lists:delete(Pid, Pool)]}}
    end;
	      
handle_cast(_Msg, S) ->
    {noreply, S}.

handle_info({'EXIT', Pid, Reason}, S=#state{sup=Sup,
					   flash=Flash}) ->
  case Pid of
      Sup ->
	  case Reason of
	      shutdown ->
		  {stop, shutdown, S};
	      _Err ->
		  redirect(S)
          end;
      Flash ->
	  restart_flash(S);
      _Others ->
	  handle_exit(Pid, Reason, S)
  end;
    
handle_info(_Info, S) ->
    {noreply, S}.

terminate(_Reason, _S) ->
    ok.

code_change(_Oldvsn, S, _Extra) ->
    {ok, S}.

%%--------internal

start_pool(Fun) ->
    lists:foldl(Fun, [], lists:seq(1, ?Size)).

acceptor(Parent, Lsock) ->
    {Tag, Sock} = gen_tcp:accept(Lsock),
    case Tag of
	error ->
	    gen_server:cast(Parent, {acceptor, {Tag, self()}}),
	    ok;
	ok ->
	    gen_server:cast(Parent, {acceptor, {Tag, self()}}),
	    loop(Sock),
	    ok
    end.


get_local_ip() ->
    {ok, Host} = inet:gethostname(),
    inet:getaddr(Host, inet).  

handle_exit(Pid, normal, S=#state{pool=Pool,
				 connect=Conn,
				 connum=Num}) ->
    case lists:member(Pid, Pool) of
	true ->
	    {noreply, S};
	false ->
	    case lists:member(Pid, Conn) of
		true ->
		    {noreply, S#state{connect=lists:delete(Pid, Conn),
				     connum=Num-1}};
		false ->
		    {noreply,S}
            end
    end;
handle_exit(Pid, _Reason, S=#state{lsock=Lsock,
				 pool=Pool,
				 connect=Conn,
				 connum=Num}) ->
    case lists:member(Pid, Pool) of
	true ->
	    Npid = spawn_link(?MODULE, acceptor, [self(), Lsock]),
	    {noreply, S#state{pool=[Npid|lists:delete(Pid, Pool)]}};
	false ->
	    case lists:member(Pid, Conn) of
		true ->
		    {noreply, S#state{connect=lists:delete(Pid, Conn),
				     connum=Num-1}};
		false ->
		    {noreply, S}
            end
     end.

redirect(S=#state{port=Port}) ->
    case global:whereis_name(netser_sup) of
	Pid when is_pid(Pid) ->
	    link(Pid),
	    gen_server:cast(Pid, {redirect, node(), Port, Pid}),
	    {noreply, S#state{sup=Pid}};
	undefined ->
	    gen_server:cast(self(), redirect),
	    {noreply, S}
    end.

restart_flash(S) ->
    Res = flash_server:start_link(self()),
    case Res of
	{ok, Pid} ->
	    {noreply, S#state{flash=Pid}};
	Err ->
	    {stop, Err, S}
    end.
	      
loop(Sock) ->
    process_flag(trap_exit, true),
    inet:setopts(Sock, [{active, true}]),
    receive
	{tcp_closed, Sock} ->
	    ok;
	%%%%login msg
	{tcp, Sock, Bin = <<0:8, Id:32, _B/binary>>} ->
	    login:login(Sock, Bin), %%%login success
	    Qid = ask_for_queue({ask_for_queue, Id}),
	    Exchanger = ask_for_exchanger(),
	    Qmref =erlang:monitor(process, Qid),
	    Emref = erlang:monitor(process, Exchanger),
	    gen_server:cast(Qid, {ask_msg, self(), Id}),
	    do_communicate(Sock, Qmref, Qid, Emref, Exchanger, Id);
	{tcp, Sock, Bin = <<2:8, Bin/binary>>} ->
	    login:do_register(Sock, Bin)	    
    end.

do_communicate(Sock, Qmref, Qid, Emref, Exchanger, Id) ->
   receive
       {tcp_closed, Sock} ->  %%sock closed
	   io:format("socket closed~n", []);
       {'DOWN', Qmref, _, _, _} ->   %%the queue is closed
	   erlang:demonitor(Qmref),
	   Nqid = ask_for_queue({ask_for_queue, Id}),
	   Qmref1 = erlang:monitor(process, Nqid),
	   do_communicate(Sock, Qmref1, Nqid, Emref, Exchanger, Id);
       {'DOWN', Emref, _, _, _} ->   %%the exchanger is closed
	   erlang:demonitor(Emref),
	   Nexchanger = ask_for_exchanger(),
	   Emref1 = erlang:monitor(process, Nexchanger),
	   do_communicate(Sock, Qmref, Qid, Emref1, Nexchanger, Id)
   after 0 ->
	 receive
	     {tcp, Sock, Bin} ->
		 case catch gen_server:call(Exchanger, {deliver, binary_to_list(Bin)}) of
		     ok ->
			 %%gen_server:cast(Qid, {ask_msg, self(), Id}),
			 do_communicate(Sock, Qmref, Qid, Emref, Exchanger, Id);
		     {'EXIT', Exchanger} ->
			 erlang:demonitor(Emref, [flush]),
			 Nexchanger = ask_for_exchanger(),
			 Emref1 = erlang:monitor(process, Nexchanger),
			 case catch gen_server:call(Nexchanger, binary_to_list(Bin)) of
			     ok ->
				 %%gen_server:cast(Qid, {ask_msg, self(), Id}),
				 do_communicate(Sock,Qmref, Qid, Emref1, Nexchanger, Id);
			     _ ->
				 io:format("exchanger error~n", []),
				 gen_tcp:close(Sock),
				 exit(normal)
                         end
                  end;
	     {Qid, Msg} ->
		 gen_tcp:send(Sock, list_to_binary(Msg)),
		 Qid ! {ack, ok},
		 gen_server:cast(Qid, {ask_msg, self(), Id})

        end
   end.   

sleep(T) ->
    receive
    after T ->
	    ok
    end.
%%%
ask_for_queue(Msg) ->
    case global:whereis_name(cache_server) of
	Pid when is_pid(Pid) ->
	    case catch gen_server:call(Pid, Msg) of
		{'EXIT', Pid} ->
		    sleep(1000),
		    ask_for_queue(Msg);
		{ok, Qid} ->
		    Qid;
		 _ ->
		    sleep(1000),
		    ask_for_queue(Msg)
            end;
	undefined ->
	    io:format("not find the cache server~n", [])
    end.
	       
ask_for_exchanger() ->
    ask_for_exchanger(3).

ask_for_exchanger(0) ->
    io:format("not find the exchanger~n", []);
ask_for_exchanger(Try) ->
    case pg2:get_members(exchanger) of
	{error, _} ->
	    ask_for_exchanger(Try-1);
	[] ->
	    ask_for_exchanger(Try-1);
	_List ->
	    pg2:get_closest_pid(exchanger)
    end.


%%%
%%login(Sock, Bin = <<_Type:8, Id:32, Pwd/binary>>) ->
%%    B = <<0:8>>,
  %%  L = binary_to_list(B),
   %% gen_tcp:send(Sock, list_to_binary(L ++ "whhbest")),
    %%exit(normal).
